<?php

namespace app\workerman\logic;

use app\common\logic\AgentLogic;
use app\common\logic\MerchantLogic;
use curl\Curlrequest;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use think\facade\Cache;
use think\facade\Config;

class WorkerMan {

    private $host;
    private $port;
    private $virtualHost;
    private $user;
    private $password;
    protected $queueName;

    function __construct() {
        $config = Config::pull('amqp');
        $this->host = $config['host'];
        $this->port = $config['port'];
        $this->virtualHost = $config['virtualHost'];
        $this->user  = $config['user'];
        $this->password  = $config['password'];
    }

    public function getConnection() {
        return new AMQPStreamConnection($this->host, $this->port, $this->user, $this->password, $this->virtualHost);
    }

    /**
     * @desc 消息队列处理1
     * @param  string $queueName 队列名称
     * @return bool
     * @throws \Exception
     */
    public function handle($queueName) {
        $connection = $this->getConnection();
        $channel = $connection->channel();

        $this->queueName = $queueName;
        $channel->queue_declare($this->queueName, false, true, false, false);
        $callback = function ($msg) {
//            echo "[Datetime: ". date('Y-m-d H:i:s'). "]\n";
//            echo "[Received: ". $msg->body . "]\n";

            $result  = $this->doJob($msg);
            if($result){
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            }
        };
        $channel->basic_qos(null, 1, null);

        // 启动队列消费者 消费者设置手动ack，在声明no_ack参数时指定false
        $channel->basic_consume($this->queueName, '', false, false, false, false, $callback);

        //echo "[Datetime: ".date('Y-m-d H:i:s') ."]\n";
        //echo "[Received: channel->callbacks:". count($channel->callbacks)."]\n";;

        while (count($channel->callbacks)) {
            try {
                $channel->wait();
                //$channel->wait(null, false, 180);
            } catch (\Exception $e) {
                echo "[Datetime: ".date('Y-m-d H:i:s') ."]\n";
                echo "[Received: Exception Message:" .$e->getMessage() ."]\n";
                $channel->close();
                $connection->close();
            }
        }
    }

    public function handleOnce($queueName){
        $connection = $this->getConnection();
        $channel = $connection->channel();

        $this->queueName = $queueName;
        $channel->queue_declare($this->queueName, false, true, false, false);
        try {
            $msg = $channel->basic_get($this->queueName);
            $result = $this->doJob($msg);
            if($result){
                $channel->basic_ack($msg->delivery_info['delivery_tag']);
            }
            $channel->close();
            $connection->close();
        }catch (\Exception $e){
            echo 'handleOnce 遇到执行错误：'.$e->getMessage()."\n";
            $channel->close();
            $connection->close();
        }

    }

    /**
     * 处理得到的 mq 消息
     * @param $msg
     * @return bool
     */
    public function doJob($msg){
        if(!$msg){
            return false;
        }
        $result = json_decode($msg->body, true);
        switch ($this->queueName) {
            case 'ORDER_NOTIFY':
                $result = $this->orderNotify($result); //通知成功返回true，销毁队列
                break;
            case 'DAIFU_NOTIFY':
                $result = $this->daifuNotify($result); //通知成功返回true，销毁队列
                break;

        }
        return $result;
    }

    public function handleBackup($queueName) {
        //创建连接
        $connection = $this->getConnection();
        //创建channel，多个channel可以共用连接
        $channel = $connection->channel();

        $this->queueName = $queueName;

        // 可能会在数据发布之前启动消费者，所以我们要确保队列存在，然后再尝试从中消费消息。
        // 创建直连的交换机
        $channel->exchange_declare('amq.direct', 'direct', false, false, false);

        // 队列创建申明 第三个参数为true时，表示队列持久化,需要和客户端一致
        $channel->queue_declare($this->queueName, false, true, false, false);

        // 交换机跟队列的绑定，
        $channel->queue_bind($this->queueName, 'amq.direct', $queueName);

        $callback = function ($msg) {
            echo "[Datetime: ". date('Y-m-d H:i:s'). "]\n";
            echo "[Received: ". $msg->body . "]\n";

            $result = json_decode($msg->body, true);
            switch ($this->queueName) {
                case 'ORDER_NOTIFY':
                    $result = $this->orderNotify($result); //通知成功返回true，销毁队列
                    break;
            }
            if ($result) {
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            }
        };
        // 启动队列消费者
        $channel->basic_qos(null, 1, null);

        // 启动队列消费者 消费者设置手动ack，在声明no_ack参数时指定false
        $channel->basic_consume($this->queueName, '', false, false, false, false, $callback);

        echo "[Received: ".date('Y-m-d H:i:s'). '$channel->callbacks:'. count($channel->callbacks)."]\n";;
        while (count($channel->callbacks)) {
            try {
                //$channel->wait();
                $channel->wait(null, false, 120);
            } catch (\Exception $e) {
                echo "[Datetime: ".date('Y-m-d H:i:s') ."]\n";
                echo "[Received: Exception Message:" .$e->getMessage() ."]\n";
                $channel->close();
                $connection->close();
            }
        }
    }

    /**
     * @desc 队列
     */
    public function redisHandle() {
        $handler = model('common/Redis', 'logic')->initialize();
        $redisQueueName = Config::get('redis.queue')['ORDER_NOTIFY'];
        $msg = $handler->lPop($redisQueueName);
        if ($msg) {
            $params = json_decode($msg, true);

            $result = $this->orderNotify($params); //通知成功返回true，销毁队列
            //重新加入队列
            if ($result === false) {
                $handler->rPush($redisQueueName, $msg);
            }
        }
    }

    /**
     * @desc 接收消息队列
     * @param string $queueName 队列名称
     * @throws \Exception
     */
    public function receiveWithProp($queueName) {
        $connection = $this->getConnection();

        $channel = $connection->channel();

        $channel->queue_declare($queueName, false, false, false, false);

        $callback = function ($msg) {
            echo ' [x] Received ', $msg->body, "\n";
            $headers = $msg->get('application_headers');
            echo ' [x] Received ', $headers->getNativeData()['delay'], "\n";
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };

        $channel->basic_consume($queueName, '', false, true, false, false, $callback);

        while (count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }

    /**
     * @desc 订单通知
     * @param array $params
     * @return bool
     * @throws \Exception
     */
    public function orderNotify($params) {
        set_time_limit(0);
        writeLog('接受的参数', 'workerManOrderNotify');
        writeLog($params, 'workerManOrderNotify');

        $lock_key = 'workerMan:orderNotify_lock_'.$params['order_no'];
        if($lock_value = Cache::has($lock_key)){
            writeLog('缓存key：'.$lock_key.'，缓存值：'.print_r($lock_value, true).'，订单[order_no:' . $params["order_no"] . '] 队列重复拦截', 'workerManOrderNotify');
            return true;
        }
        Cache::set($lock_key, 'repeat', 45);

        //查询订单信息
        $orderInfo = model('Orders')
            ->alias('o')
            ->field('o.*,oc.order_id,oc.notify_status,oc.notify_time,oc.retmsg,oc.times')
            ->where('o.order_no','=',$params["order_no"])
            ->leftJoin('OrdersCallback oc', 'oc.order_id = o.id')
//            ->lock(true)
            ->find();

        if(!$orderInfo) {
            writeLog('订单[order_no:' . $params["order_no"] . ']信息不存在', 'workerManOrderNotify');
            return true;
        }

        //订单通知次数
        if($orderInfo['times'] >= 3) {
            writeLog('订单[order_no:' . $orderInfo["order_no"] . ']通知次数超过3次', 'workerManOrderNotify');
            //将订单修改为通知失败状态  通知状态 0  未通知 1 通知成功 2 通知失败
            $ordersCallback = model("OrdersCallback")->where(["order_id" => $orderInfo['id']])->update([
                "notify_status" => 2,
                "notify_time"   => date("Y-m-d H:i:s"),
                "retmsg"        => "订单[order_no:". $orderInfo['order_no'] ."]订单通知次数超过3次～"
            ]);
            writeLog('订单[order_no:' . $orderInfo["order_no"] . ']通知失败更新结果：'.$ordersCallback, 'workerManOrderNotify');
            return true;
        }

        //是否已接收到上游成功通知  0 未接收  1 已接收
        if ($orderInfo['fettle_status'] == 0) {
            writeLog('订单[order_no:' . $orderInfo["order_no"] . ']未接收到上游通知', 'workerManOrderNotify');
            return true;
        }

        //通知状态 0  未通知 1 通知成功 2 通知失败
        if ($orderInfo["notify_status"] == 1) {
            writeLog('订单[order_no:' . $orderInfo["order_no"] . ']订单已通知成功', 'workerManOrderNotify');
            return true;
        }

        //支付状态  1 未支付  2 已支付  3 订单超时  4 冻结订单
        if ($orderInfo["pay_status"] != 2) {
            writeLog('订单[order_no:' . $orderInfo["order_no"] . ']订单未支付', 'workerManOrderNotify');
            return true;
        }

        $channel = model("Channel")->where(["channel_id" => $orderInfo['channel_id']])->find();
        if ($channel['channel_switch'] != 1) {
            writeLog('通道[channel_id:' . $orderInfo['channel_id'] . ']已关闭', 'workerManOrderNotify');
            return true;
        }

        // 判断商户状态是否正常
        $user = model("User")->field("user_id,mch_id,mch_key")
            ->where(["user_id" => $orderInfo['user_id'], "status" => 1])
            ->find();
        if (!$user) {
            writeLog('商户[user_id:' . $params['user_id'] . ']已关闭', 'workerManOrderNotify');
            return true;
        }

        // 查询商户对应通道是否开启与费率
        $user_rate = model("Rate")
            ->where(["user_id" => $params['user_id'], "channel_id" => $orderInfo['channel_id'], "is_state" => 1])
            ->value("rate");
        if (!isset($user_rate) && empty($user_rate)) {
            writeLog('商户:[' . $orderInfo['user_id'] . ']未开启通道汇率', 'workerManOrderNotify');
            return true;
        }

        $payTime= $orderInfo['update_time'] ?: date('Y-m-d H:i:s');
        // 异步数组
        $postData = [
            "mch_id"      => $user["mch_id"],                               // 商户号
            "order_no"    => $orderInfo["order_no"],                        // 商户订单号
            "pay_status"  => $orderInfo["pay_status"],                       // 支付状态
            "usdt_amount" => round($orderInfo["usdt_amount"], 4),  // usdt数量
            "usdt_rate"   => round($orderInfo["usdt_rate"], 4),    // usdt汇率
            "pay_amount"  => round($orderInfo["amount"], 4),       // 支付金额
            "pay_time"    => strtotime($payTime)                            // 支付时间
        ];
        //创建签名
        $postData['sign'] = build_sign($postData,$user['mch_key']);
        $httpData = [
            "errorCode" => "200",
            "message"   => "操作成功",
            "data"      => $postData
        ];
        writeLog('post商户参数', 'workerManOrderNotify');
        writeLog($httpData, 'workerManOrderNotify');

        // 通知次数+1
        model("OrdersCallback")->where(["order_id" => $params['id']])->setInc("times");

        //检测通知地址是否是https或者http协议地址
        if(!$params['async_url']) {
            writeLog('通知地址有误:' . $params['async_url'], 'workerManDaifuNotify');
            return true;
        }

        $responseJson = Curlrequest::curlPostJson($params['async_url'], json_encode($httpData), 30);
        /**
         * 28为请求超时的错误码
         * CURLE_OPERATION_TIMEDOUT (28) – 操作超时。 已达到根据相应情况指定的超时时间。
         */
        if ($responseJson == 28) {
            writeLog("curl操作超时:CURLE_OPERATION_TIMEDOUT (28)", 'workerManOrderNotify');
            return false;
        }
        writeLog('请求响应结果,curlResult：' . $responseJson, 'workerManOrderNotify');


        if (strcmp(trim($responseJson, '"'), 'success') == 0) {
            //todo 成功后逻辑处理
        } else {
            writeLog('订单[order_id:' . $params['id'] . ']请求响应结果不为success', 'workerManOrderNotify');
            // 接收响应数据
            model("OrdersCallback")->where(["order_id" => $params['id']])->update([
                "response_message" => $responseJson
            ]);
            return false;
        }
    }

    /**
     * @desc 代付通知
     * @param $params
     * @return bool
     * @throws \Exception
     */
    public function daifuNotify($params) {
        set_time_limit(0);
        writeLog('接受的参数', 'workerManDaifuNotify');
        writeLog($params, 'workerManDaifuNotify');

        //查询订单信息
        $daifuInfo = model('UserDaifu')
            ->where('daifu_id', '=', $params['daifu_id'])
            ->find();

        if(!$daifuInfo) {
            writeLog('代付订单[order_no:' . $params["order_no"] . ']信息不存在', 'workerManDaifuNotify');
            return true;
        }

        //通知状态{0-未通知 1-通知成功 2-通知失败}
        if ($daifuInfo["notify_status"] == 1) {
            writeLog('订单[order_no:' . $params["order_no"] . ']订单已通知成功,无需再次通知', 'workerManDaifuNotify');
            return true;
        }

        //订单通知次数
        if($daifuInfo['times'] >= 3) {
            writeLog('订单[order_no:' . $daifuInfo["order_no"] . ']通知次数超过3次', 'workerManDaifuNotify');
            //通知状态{0-未通知 1-通知成功 2-通知失败}
            $notifyResult = model("UserDaifu")
                ->where('daifu_id', '=', $daifuInfo['daifu_id'])
                ->update([
                    "notify_status" => 2,
                    "notify_time"   => date("Y-m-d H:i:s")
                ]);
            writeLog('订单[order_no:' . $daifuInfo["order_no"] . ']通知失败更新结果：'.$notifyResult, 'workerManDaifuNotify');
            return true;
        }

        //状态{0-待审核 1-处理 2-审核通过 3-审核拒绝}
        if (in_array($daifuInfo["status"], [0, 1])) {
            writeLog('订单[order_no:' . $daifuInfo["order_no"] . ']订单未审核', 'workerManDaifuNotify');
            return true;
        }

        // 判断商户状态是否正常
        $user = model("User")
            ->where(["user_id" => $daifuInfo['user_id'], "status" => 1])
            ->find();
        if (!$user) {
            writeLog('商户[user_id:' . $daifuInfo['user_id'] . ']已关闭', 'workerManDaifuNotify');
            return true;
        }

        // 异步数组
        $postData = [
            "mch_id"           => $user["mch_id"],                            // 商户号
            "order_no"         => $daifuInfo["order_no"],                     // 系统订单号
            "out_order_no"     => $daifuInfo["out_order_no"],                 // 商户订单号
            "bank_name"        => $daifuInfo["bank_name"],                    // 银行名称
            "bank_branch"      => $daifuInfo["bank_branch"],                  // 支行信息
            "account_name"     => $daifuInfo["account_name"],                 // 户名
            "account_no"       => $daifuInfo["account_no"],                   // 卡号
            "amount"           => $daifuInfo["amount"],                       // 代付金额
            "service_fee"      => $daifuInfo["service_fee"],                  // 代付手续费
            "transfer_receipt" => $daifuInfo["transfer_receipt"],             // 转账凭证
            "status"           => $daifuInfo["status"],                       // 状态{0-待审核 1-处理中 2-审核通过 3-审核拒绝}
            "confirm_time"     => strtotime($daifuInfo['confirm_time']),      // 审核时间
            "create_time"      => strtotime($daifuInfo['create_time'])        // 发起时间
        ];
        //创建签名
        $postData['sign'] = build_sign($postData, $user['mch_key']);
        $httpData = [
            "errorCode" => "200",
            "message"   => "操作成功",
            "data"      => $postData
        ];
        writeLog('post商户参数', 'workerManDaifuNotify');
        writeLog($httpData, 'workerManDaifuNotify');

        //检测通知地址是否是https或者http协议地址，如果有就通知，否则不予通知让商户自己定时轮询
        if($daifuInfo['notify_url']) {
            writeLog('开始通知商户', 'workerManDaifuNotify');
            $responseJson = Curlrequest::curlPostJson($daifuInfo['notify_url'], json_encode($httpData), 30);
            /**
             * 28为请求超时的错误码
             * CURLE_OPERATION_TIMEDOUT (28) – 操作超时。 已达到根据相应情况指定的超时时间。
             */
            if ($responseJson == 28) {
                writeLog("curl操作超时:CURLE_OPERATION_TIMEDOUT (28)", 'workerManDaifuNotify');
                return false;
            }
            writeLog('请求响应结果,curlResult：' . $responseJson, 'workerManDaifuNotify');

            if (strcmp($responseJson, 'success') == 0) {
                $notifyResult = model("UserDaifu")
                    ->where('daifu_id', '=', $daifuInfo['daifu_id'])
                    ->update([
                        "notify_status" => 1,
                        "notify_time"   => date("Y-m-d H:i:s"),
                        "notify_result" => '请求响应结果,curlResult：' . $responseJson
                    ]);
                writeLog('订单[order_no:' . $daifuInfo['order_no'] . ']修改状态:' . $notifyResult, 'workerManDaifuNotify');
                return true;
            } else {
                writeLog('订单[order_no:' . $daifuInfo['order_no'] . ']请求响应结果不为success', 'workerManDaifuNotify');
                // 接收响应数据
                $notifyResult = model("UserDaifu")
                    ->where('daifu_id', '=', $daifuInfo['daifu_id'])
                    ->update([
                        "notify_status" => 2,
                        "notify_time"   => date("Y-m-d H:i:s"),
                        'notify_result' => '请求响应结果,curlResult：' . $responseJson
                    ]);
                writeLog('订单[order_no:' . $daifuInfo["order_no"] . ']通知失败更新结果：'.$notifyResult, 'workerManDaifuNotify');
                return false;
            }
        }
        return true;
    }
}
